package com.bw.dwd;

import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.util.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Random;

/**
 * @BelongsProject: work_new_order
 * @BelongsPackage: com.bw.dwd
 * @Author: wuxiaopeng
 * @CreateTime: 2025-07-08  22:31
 * @Description: TODO
 * @Version: 1.0
 */
public class ods_to_dwd {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//      读取Kafka数据
        DataStreamSource<String> file = KafkaUtil.getKafkaSource(env, "new_work_order_ods_all_data", "asdfa");

//      创建测流放到Kafka
        OutputTag<String> infoTag = new OutputTag<String>("infoTag"){};
        OutputTag<String> detailTag = new OutputTag<String>("detailTag"){};
        OutputTag<String> cartTag = new OutputTag<String>("cartTag") {};
        OutputTag<String> paymentTag = new OutputTag<String>("paymentTag") {};

//      筛选出对应测流的数据
        SingleOutputStreamOperator<Object> stream = file.process(new ProcessFunction<String, Object>() {
            @Override
            public void processElement(String s, ProcessFunction<String, Object>.Context context, Collector<Object> collector) throws Exception {
                JSONObject all = JSON.parseObject(s);
                String table = all.getJSONObject("source").getString("table");
                if ("order_info".equals(table)) {
                    all.getJSONObject("after").put("shop_id",RandomUtil.randomInt(1111,1117));
                    context.output(infoTag, all.getString("after"));
                }
                if ("order_detail".equals(table)) {
                    context.output(detailTag, all.getString("after"));
                }
                if ("cart_info".equals(table)) {
                    context.output(cartTag, all.getString("after"));
                }
                if ("payment_info".equals(table)) {
                    context.output(paymentTag, all.getString("after"));
                }
            }
        });

        SideOutputDataStream<String> info = stream.getSideOutput(infoTag);
        SideOutputDataStream<String> detail = stream.getSideOutput(detailTag);
        SideOutputDataStream<String> cart = stream.getSideOutput(cartTag);
        SideOutputDataStream<String> payment = stream.getSideOutput(paymentTag);

        info.print();

//      发到Kafka

//        info.addSink(KafkaUtil.getKafkaSink("new_work_order_dwd_info"));
//        detail.addSink(KafkaUtil.getKafkaSink("new_work_order_dwd_detail"));
//        cart.addSink(KafkaUtil.getKafkaSink("new_work_order_dwd_cart"));
//        payment.addSink(KafkaUtil.getKafkaSink("new_work_order_dwd_payment"));


        env.execute("work_order");
    }
}
